package com.happy3w.playground.thread;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

public class Worker<T> implements Runnable {
    private final LinkedBlockingQueue<T> queue;
    private final AtomicInteger runningWorkerCount;
    private final Consumer<T> consumer;
    private final T endFlag;

    public Worker(Consumer<T> consumer, T endFlag, LinkedBlockingQueue<T> queue, AtomicInteger runningWorkerCount) {
        this.queue = queue;
        this.runningWorkerCount = runningWorkerCount;
        this.consumer = consumer;
        this.endFlag = endFlag;
    }

    @Override
    public void run() {
        while (true) {
            try {
                runningWorkerCount.incrementAndGet();
                T data = queue.take();
                if (data == endFlag) {
                    break;
                }

                consumer.accept(data);
            } catch (InterruptedException e) {
                break;
            } catch (Exception e) {
                continue;
            } finally {
                runningWorkerCount.decrementAndGet();
            }
        }
    }
}
